package com.kingsoft.dc.khaos.module.spark.preprocess.specific

import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.innertype.Schema
import com.kingsoft.dc.khaos.metadata.Dependency
import com.kingsoft.dc.khaos.module.spark.preprocess.transform.TransformStrategy
import com.kingsoft.dc.khaos.util.{KhaosConstants, Logging}
import org.apache.spark.sql.DataFrame
import org.apache.spark.util.LongAccumulator


/**
  * create by haorenhui on 2020/06/28.
  */
class CollectMetrics extends TransformStrategy with Logging {

  override def exec(kc: KhaosContext,
                    module_id: String,
                    config: String,
                    dependences: Seq[Dependency],
                    targets: Seq[Dependency]): Seq[(String, DataFrame)] = {
    val abnormalAccumulator: LongAccumulator = kc._valuesChannel.getValues[LongAccumulator](KhaosConstants.KHAOS_ABNORMALACCUMULATOR)
    log.info(s"khaos  metrics: ignore abnormal data num is $abnormalAccumulator")

    val df: DataFrame = null
    addResult(Dependency(), df)
  }

  override def schema(kc: KhaosContext,
                      module_id: String,
                      config: String,
                      dependences: Seq[Dependency]): Any = {
    new Schema(Nil)
  }
}
